package ConsumerTest;

import java.io.FileInputStream;
import java.io.IOException;
import java.util.Arrays;
import java.util.Properties;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import ty.pub.Error;
public class ErrorConsumer extends Thread{

	static KafkaConsumer<String, Error> consumer;
	
	
	public ErrorConsumer(Properties props){
		consumer = new KafkaConsumer<String, Error>(props);
		consumer.subscribe(Arrays.asList(props.getProperty("error.topic")));
	}
	
	public static void main(String args[]) throws IOException {
		// TODO Auto-generated method stub
		Properties props = new Properties();
		FileInputStream in = new FileInputStream("error.props");
		props.load(in);
		in.close();
		new ErrorConsumer(props);
		consumer.poll(100);
		consumer.seekToBeginning(consumer.assignment());
		while (true) {
			ConsumerRecords<String, Error> records = consumer.poll(100);
			for (ConsumerRecord<String, Error> record : records){
				System.out.println(record.value().getRawDataId());
			}
		}
	}
}
